package com.shujia.hbaseapi;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.List;

public class HbaseFilterComparatorDemo {
    private HConnection conn;
    private HBaseAdmin hAdmin;

    @Before
    public void connect() {
        try {
            //1、获取Hadoop的相关配置环境
            Configuration conf = new Configuration();

            //2、获取zookeeper的配置
            conf.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");
            //获取与Hbase的连接，这个连接是将来可以用户获取hbase表的
            conn = HConnectionManager.createConnection(conf);

            //将来我们要对表做DDL相关操作，而对表的操作在hbase架构中是有HMaster
            hAdmin = new HBaseAdmin(conf);

            System.out.println("建立连接成功:" + conn + ", HMaster获取成功：" + hAdmin);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 行键过滤器
     * 通过RowFilter与BinaryComparator过滤比rowKey 1500100010小的所有值出来
     */
    @Test
    public void RowFilter1() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            BinaryComparator binaryComparator = new BinaryComparator("1500100010".getBytes());

            //创建一个行键过滤器的对象
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.LESS, binaryComparator);

            Scan scan = new Scan();
            scan.setFilter(rowFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列簇过滤器案例1：通过FamilyFilter与SubstringComparator查询列簇名包含in的所有列簇下面的数据
     */
    @Test
    public void FamilyFilter1() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建一个比较器对象
            //只要列簇名中包含了in，就把该列簇下的所有列查询出来
            SubstringComparator substringComparator = new SubstringComparator("nf");  //info

            //创建列簇过滤器
            FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, substringComparator);

            Scan scan = new Scan();
            scan.setFilter(familyFilter);

            //获取数据
            ResultScanner scanner = students.getScanner(scan);
            print(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列簇过滤器案例2：通过FamilyFilter与 BinaryPrefixComparator 过滤出列簇以i开头的列簇下的所有数据
     */
    @Test
    public void FamilyFilter2() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("i".getBytes());

            //创建列簇过滤器
            FamilyFilter familyFilter = new FamilyFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            Scan scan = new Scan();
            scan.setFilter(familyFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列过滤器案例1：通过QualifierFilter与SubstringComparator查询列名包含ge的列的值
     */
    @Test
    public void QualifierFilter1() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建包含比较器
            //age
            //gender
            SubstringComparator substringComparator = new SubstringComparator("ge");

            //创建一个列过滤器
            QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, substringComparator);

            Scan scan = new Scan();
            scan.setFilter(qualifierFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    /**
     * 列过滤器案例2：通过QualifierFilter与SubstringComparator查询列名包含ge的列的值
     */
    @Test
    public void QualifierFilter2() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            SubstringComparator substringComparator = new SubstringComparator("am");

            //创建列过滤器
            QualifierFilter qualifierFilter = new QualifierFilter(CompareFilter.CompareOp.EQUAL, substringComparator);

            Scan scan = new Scan();
            scan.setFilter(qualifierFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 列值过滤器案例1:通过ValueFilter与BinaryPrefixComparator过滤出所有的cell中值以 "张" 开头的学生
     */
    @Test
    public void ValueFilter1() {
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建前缀比较器
            BinaryPrefixComparator binaryPrefixComparator = new BinaryPrefixComparator("张".getBytes());

            //创建列值过滤器的对象
            ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, binaryPrefixComparator);

            Scan scan = new Scan();
            scan.setFilter(valueFilter);

            ResultScanner scanner = students.getScanner(scan);

            //因为ResultScanner类继承了迭代器
            //使用增强for循环遍历
//            for (Result rs : scanner) {
//                String id = Bytes.toString(rs.getRow());
//                System.out.println("当前行的rowkey为：" + id);
//                //继续增强for循环得到每一行中的每一个单元格（列）
//                //获取一行中的所有单元格
//                for (Cell cell : rs.listCells()) {
//                    //获取该单元格属于的列簇
//                    String family = Bytes.toString(CellUtil.cloneFamily(cell));
//                    //获取该单元格的列名
//                    String colName = Bytes.toString(CellUtil.cloneQualifier(cell));
//                    //获取该单元格的列值
//                    String value = Bytes.toString(CellUtil.cloneValue(cell));
//                    System.out.println(family + ":" + colName + "的值为：" + value);
//                }
//            }

            print(scanner);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     *      列值过滤器案例2：> 过滤出文科的学生，只会返回以文科开头的数据列，其他列的数据不符合条件，不会返回
     */
    @Test
    public void ValueFilter12(){
        try {
            //获取表的实例
            HTableInterface students = conn.getTable("students");

            //创建正则比较器
            RegexStringComparator regexStringComparator = new RegexStringComparator("^文科.*");

            //创建列值过滤器
            ValueFilter valueFilter = new ValueFilter(CompareFilter.CompareOp.EQUAL, regexStringComparator);

            Scan scan = new Scan();
            scan.setFilter(valueFilter);

            ResultScanner scanner = students.getScanner(scan);
            print(scanner);


        } catch (IOException e) {
            e.printStackTrace();
        }
    }



    /**
     * 专门用来打印数据的方法
     */
    public void print(ResultScanner scanner) throws IOException {
        Result rs = null;
        while ((rs = scanner.next()) != null) {
            String id = Bytes.toString(rs.getRow());
            List<Cell> cells = rs.listCells();
            System.out.print("id:" + id);
            System.out.print("\t");
            for (Cell cell : cells) {
//                String s = Bytes.toString(cell.getValue());
                String col = Bytes.toString(CellUtil.cloneQualifier(cell));
                String s = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.print(col + ":" + s);
                System.out.print("\t");
            }
            System.out.println();

//            String name = Bytes.toString(rs.getValue("info".getBytes(), "name".getBytes()));
//            String age = Bytes.toString(rs.getValue("info".getBytes(), "age".getBytes()));
//            String gender = Bytes.toString(rs.getValue("info".getBytes(), "gender".getBytes()));
//            String clazz = Bytes.toString(rs.getValue("info".getBytes(), "clazz".getBytes()));
//            System.out.println("学号：" + id + ",姓名：" + name + ",年龄：" + age + ",性别：" + gender + ",班级：" + clazz);

        }
    }


    @After
    public void close() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("conn连接已经关闭.....");
        }

        if (hAdmin != null) {
            try {
                hAdmin.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
            System.out.println("HMaster已经关闭......");
        }
    }

}
